/**
* @file stream.cpp
*
* Copyright (c) Huawei Technologies Co., Ltd. 2019-2020. All Rights reserved.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
*/

#include "acl/acl_rt.h"

#include "runtime/stream.h"
#include "framework/executor/ge_executor.h"
#include "register/stream_manage_func_registry.h"

#include "log_inner.h"
#include "single_op/executor/stream_executor.h"
#include "model/acl_resource_manager.h"
#include "error_codes_inner.h"
#include "toolchain/profiling_manager.h"
#include "toolchain/resource_statistics.h"

aclError aclrtCreateStream(aclrtStream *stream)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtCreateStream);
    ACL_ADD_APPLY_TOTAL_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    ACL_LOG_INFO("start to execute aclrtCreateStream");
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(stream);

    rtStream_t rtStream = nullptr;
    const rtError_t rtErr = rtStreamCreate(&rtStream, static_cast<int32_t>(RT_STREAM_PRIORITY_DEFAULT));
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("create stream failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }

    *stream = static_cast<aclrtStream>(rtStream);
    ACL_ADD_APPLY_SUCCESS_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    return ACL_SUCCESS;
}

aclError aclrtCreateStreamWithConfig(aclrtStream *stream, uint32_t priority, uint32_t flag)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtCreateStreamWithConfig);
    ACL_ADD_APPLY_TOTAL_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    ACL_LOG_INFO("start to execute aclrtCreateStreamWithConfig with priority:%u, flag:%u", priority, flag);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(stream);

    uint32_t streamFlag = 0U;
    if ((flag & ACL_STREAM_FAST_LAUNCH) != 0U) {
        streamFlag |= RT_STREAM_FAST_LAUNCH;
    }
    if ((flag & ACL_STREAM_FAST_SYNC) != 0U) {
        streamFlag |= RT_STREAM_FAST_SYNC;
    }

    rtStream_t rtStream = nullptr;
    const rtError_t rtErr = rtStreamCreateWithFlagsExternal(&rtStream, priority, streamFlag);
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("create stream failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }

    *stream = static_cast<aclrtStream>(rtStream);
    ACL_ADD_APPLY_SUCCESS_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    return ACL_SUCCESS;
}

aclError aclrtDestroyStream(aclrtStream stream)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtDestroyStream);
    ACL_ADD_RELEASE_TOTAL_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    ACL_LOG_INFO("start to execute aclrtDestroyStream");
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(stream);

    (void) ge::GeExecutor::ReleaseSingleOpResource(stream);
    (void) acl::Executors::RemoveExecutor(nullptr, stream);
    (void) acl::AclResourceManager::GetInstance().CleanRT2Executor(stream);
    acl::AclResourceManager::GetInstance().CleanAllocators(stream);
    ACL_LOG_INFO("release singleop success");
    ge::MngResourceHandle streamHandle = {.stream = static_cast<rtStream_t>(stream)};
    (void)ge::StreamMngFuncRegistry::GetInstance().TryCallStreamMngFunc(
        ge::StreamMngFuncType::ACLNN_STREAM_CALLBACK, ge::MngActionType::DESTROY_STREAM, streamHandle);
    ACL_LOG_INFO("try to call stream manage function success, action type is DESTROY_STREAM.");
    const rtError_t rtErr = rtStreamDestroy(static_cast<rtStream_t>(stream));
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("destroy stream failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }
    ACL_LOG_INFO("aclrtDestroyStream success");
    ACL_ADD_RELEASE_SUCCESS_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    return ACL_SUCCESS;
}

aclError aclrtDestroyStreamForce(aclrtStream stream)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtDestroyStreamForce);
    ACL_ADD_RELEASE_TOTAL_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    ACL_LOG_INFO("start to execute aclrtDestroyStreamForce");
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(stream);

    (void) ge::GeExecutor::ReleaseSingleOpResource(stream);
    (void) acl::Executors::RemoveExecutor(nullptr, stream);
    (void) acl::AclResourceManager::GetInstance().CleanRT2Executor(stream);
    acl::AclResourceManager::GetInstance().CleanAllocators(stream);
    ACL_LOG_INFO("release singleop success");
    ge::MngResourceHandle streamHandle = {.stream = static_cast<rtStream_t>(stream)};
    (void)ge::StreamMngFuncRegistry::GetInstance().TryCallStreamMngFunc(
        ge::StreamMngFuncType::ACLNN_STREAM_CALLBACK, ge::MngActionType::DESTROY_STREAM, streamHandle);
    ACL_LOG_INFO("try to call stream manage function success, action type is DESTROY_STREAM.");
    const rtError_t rtErr = rtStreamDestroyForce(static_cast<rtStream_t>(stream));
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("destroy stream force failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }
    ACL_LOG_INFO("aclrtDestroyStreamForce success");
    ACL_ADD_RELEASE_SUCCESS_COUNT(acl::ACL_STATISTICS_CREATE_DESTROY_STREAM);
    return ACL_SUCCESS;
}

aclError aclrtSynchronizeStream(aclrtStream stream)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtSynchronizeStream);
    ACL_LOG_INFO("start to execute aclrtSynchronizeStream");

    const rtError_t rtErr = rtStreamSynchronize(static_cast<rtStream_t>(stream));
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("synchronize stream failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }

    ACL_LOG_INFO("Synchronize stream success");
    return ACL_SUCCESS;
}

aclError aclrtSynchronizeStreamWithTimeout(aclrtStream stream, int32_t timeout)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtSynchronizeStreamWithTimeout);
    ACL_LOG_INFO("start to execute aclrtSynchronizeStreamWithTimeout, timeout = %d", timeout);
    constexpr int32_t default_timeout = -1;
    if (timeout < default_timeout) {
        ACL_LOG_CALL_ERROR("the timeout of synchronize stream is invalid");
        return ACL_ERROR_RT_PARAM_INVALID;
    }
    const rtError_t rtErr = rtStreamSynchronizeWithTimeout(static_cast<rtStream_t>(stream), timeout);
    if (rtErr == ACL_ERROR_RT_STREAM_SYNC_TIMEOUT) {
        ACL_LOG_CALL_ERROR("synchronize stream timeout, timeout = %d", timeout);
        return ACL_ERROR_RT_STREAM_SYNC_TIMEOUT;
    } else if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("synchronize stream failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }

    ACL_LOG_INFO("Synchronize stream with timeout success");
    return ACL_SUCCESS;
}

aclError aclrtStreamQuery(const aclrtStream stream, aclrtStreamStatus *status)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtStreamQuery);
    ACL_LOG_INFO("start to execute aclrtStreamQuery");
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(status);

    const rtError_t rtErr = rtStreamQuery(static_cast<rtStream_t>(stream));
    if (rtErr == RT_ERROR_NONE) {
        *status = ACL_STREAM_STATUS_COMPLETE;
    } else if (rtErr == ACL_ERROR_RT_STREAM_NOT_COMPLETE) {
        *status = ACL_STREAM_STATUS_NOT_READY;
    } else {
        ACL_LOG_CALL_ERROR("stream query failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }

    ACL_LOG_INFO("successfully execute aclrtStreamQuery");
    return ACL_SUCCESS;
}

aclError aclrtStreamWaitEvent(aclrtStream stream, aclrtEvent event)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtStreamWaitEvent);
    ACL_LOG_INFO("start to execute aclrtStreamWaitEvent");
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(event);

    const rtError_t rtErr = rtStreamWaitEvent(static_cast<rtStream_t>(stream), static_cast<rtEvent_t>(event));
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("stream wait event failed, runtime result = %d", static_cast<int32_t>(rtErr));
        return ACL_GET_ERRCODE_RTS(rtErr);
    }

    ACL_LOG_INFO("stream wait event success");
    return ACL_SUCCESS;
}

aclError aclrtSetStreamFailureMode(const aclrtStream stream, const uint64_t mode)
{
    ACL_PROFILING_REG(acl::AclProfType::AclrtSetStreamFailureMode);
    ACL_LOG_INFO("start to execute aclrtSetStreamFailureMode, mode is %lu", mode);
    const rtError_t rtErr = rtStreamSetMode(static_cast<rtStream_t>(stream), mode);
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("rtSetStreamFailureMode failed, runtime result = %d.", rtErr);
        return ACL_GET_ERRCODE_RTS(rtErr);
    }
    ACL_LOG_INFO("successfully execute aclrtSetStreamFailureMode, mode is %lu", mode);
    return ACL_SUCCESS;
}

aclError aclrtGetStreamOverflowSwitch(aclrtStream stream, uint32_t *flag)
{
    ACL_LOG_INFO("start to execute aclrtGetStreamOverflowSwitch");
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(flag);
    const rtError_t rtErr = rtGetStreamOverflowSwitch(static_cast<rtStream_t>(stream), flag);
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("rtGetStreamOverflowSwitch failed, runtime result = %d.", rtErr);
        return ACL_GET_ERRCODE_RTS(rtErr);
    }
    ACL_LOG_INFO("successfully execute aclrtGetStreamOverflowSwitch, flag is %d.", *flag);
    return ACL_SUCCESS;
}

aclError aclrtSetStreamOverflowSwitch(aclrtStream stream, uint32_t flag)
{
    ACL_LOG_INFO("start to execute aclrtSetStreamOverflowSwitch, flag is %u.", flag);
    if ((flag != 0U) && ((flag != 1U))) {
        ACL_LOG_ERROR("flag must be 1 or 0, but current value is %u", flag);
        return ACL_ERROR_INVALID_PARAM;
    }
    const rtError_t rtErr = rtSetStreamOverflowSwitch(static_cast<rtStream_t>(stream), flag);
    if (rtErr != RT_ERROR_NONE) {
        ACL_LOG_CALL_ERROR("rtSetStreamOverflowSwitch failed, runtime result = %d.", rtErr);
        return ACL_GET_ERRCODE_RTS(rtErr);
    }
    ACL_LOG_INFO("successfully execute rtSetStreamOverflowSwitch, flag is %u.", flag);
    return ACL_SUCCESS;
}
